package com.azh.server.tx;

import java.io.IOException;
import java.io.RandomAccessFile;
import java.nio.ByteBuffer;
import java.nio.channels.FileChannel;
import java.nio.file.FileSystemException;
import java.util.Arrays;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;

import com.azh.server.util.ErrorUtil;
import com.azh.server.util.ExceptionUtils;
import com.azh.server.util.Parser;

/**
 * @author zhaozhenhang <zhaozhenhang@kuaishou.com>
 * Created on 2022-04-24
 */
public class TxManagerImpl implements TxManager {

    // 事物文件 文件头长度
    public static final int TX_ID_HEADER_LENGTH = 8;

    // 每个事物的占用长度
    private static final byte TX_ID_FIELD_SIZE = 1;

    // 事物三种状态
    // 正在进行
    private static final byte FIELD_TX_ACTIVE = 0;

    // 提交
    private static final byte FIELD_TX_COMMITTED = 1;

    // 回滚
    private static final byte FIELD_TX_ABORTED = 2;

    // 超级事物，该事物一直处于committed状态
    public static final long SUPER_TX_ID = 0;

    // 事物文件后缀
    public static String TX_ID_SUFFIX = ".aid";

    private RandomAccessFile file;
    private FileChannel channel;
    private long txIdCounter;
    // 采用reentrantLock实现，后续考虑无锁实现
    private Lock counterLock;

    /**
     * 使用nio的fileChannel实现，后续考虑利用netty的实现
     * @param file
     * @param channel
     */
    public TxManagerImpl(RandomAccessFile file, FileChannel channel) {
        this.file = file;
        this.channel = channel;
        counterLock = new ReentrantLock();
        checkTxIdCounter();
    }

    /**
     * 校验文件是否合法，类似class 魔法数的概念，用文件头的数据校验
     * 不同的是class文件是魔法书判断是否合法
     * 这儿是通过头部长度反推文件理论长度，和实际长度比较，如果不同就是不合法
     * @deprecated TX_ID_HEADER_LENGTH 通过该长度计算file长度
     */
    private void checkTxIdCounter() {
        long fileLens = 0;
        try {
            fileLens = file.length();
        } catch (IOException e) {
            ExceptionUtils.getMessage(e);
        }
        if (fileLens < TX_ID_HEADER_LENGTH) {
            ErrorUtil.error(new FileSystemException("文件格式有问题"));
        }
        ByteBuffer buffer = ByteBuffer.allocate(TX_ID_HEADER_LENGTH);
        try {
            channel.position(0);
            channel.read(buffer);
        } catch (IOException e) {
            ExceptionUtils.getMessage(e);
        }
        this.txIdCounter = Parser.parseLong(buffer.array());
        long end = getTxIdPosition(this.txIdCounter + 1);
        if (end != fileLens) {
            ErrorUtil.error(new FileSystemException("文件格式有问题，或者可能文件内容错误"));
        }
    }

    /**
     * 根据事物id获取当前事物在文件中的位置
     * @param txId
     * @return
     */
    private long getTxIdPosition(long txId) {
        // todo 该方式是事物长度都一样，有个不好的地方就是有内存浪费，后续修改成数组记录事物长度的方式
        return TX_ID_HEADER_LENGTH + (txId - 1) * TX_ID_FIELD_SIZE;
    }

    /**
     * 更新事物txId的状态
     * @param txId
     * @param status
     */
    private void updateTxStatus(long txId, byte status) {
        long offset = getTxIdPosition(txId);
        byte[] data = new byte[TX_ID_FIELD_SIZE];
        data[0] = status;
        ByteBuffer buffer = ByteBuffer.wrap(data);
        try {
            channel.position(offset);
            channel.write(buffer);
        } catch (IOException e) {
            ExceptionUtils.getMessage(e);
        }
        try {
            // 是否将这个元数据（事物状态写到磁盘上面）
            channel.force(false);
        } catch (IOException e) {
            ExceptionUtils.getMessage(e);
        }
    }

    /**
     * 增加事物id，这儿要保证原子性
     */
    private void incrTxIdCounter() {
        this.txIdCounter++;
        ByteBuffer buffer = ByteBuffer.wrap(Parser.longToByte(this.txIdCounter));
        try {
            channel.position(0);
            channel.write(buffer);
        } catch (IOException e) {
            ExceptionUtils.getMessage(e);
        }
        try {
            // 是否将这个元数据（事物id写到磁盘上面）
            channel.force(false);
        } catch (IOException e) {
            ExceptionUtils.getMessage(e);
        }
    }

    /**
     * 判断事物是否处于status状态
     * @param txId
     * @param status
     * @return
     */
    private boolean checkTrxId(long txId, byte status) {
        long offset = getTxIdPosition(txId);
        ByteBuffer buffer = ByteBuffer.wrap(new byte[TX_ID_FIELD_SIZE]);
        try {
            channel.position(offset);
            channel.read(buffer);
        } catch (IOException e) {
            ExceptionUtils.getMessage(e);
        }
        return buffer.array()[0] == status;
    }

    @Override
    public long begin() {
        counterLock.lock();
        try {
            long txId = this.txIdCounter + 1;
            updateTxStatus(txId, FIELD_TX_ACTIVE);
            incrTxIdCounter();
            return this.txIdCounter;
        }finally {
            counterLock.unlock();
        }
    }

    @Override
    public void commit(long txId) {
        updateTxStatus(txId, FIELD_TX_COMMITTED);
    }

    @Override
    public void abort(long txId) {
        updateTxStatus(txId, FIELD_TX_ABORTED);
    }

    @Override
    public boolean isActive(long txId) {
        if (txId == SUPER_TX_ID) {
            return false;
        }
        return checkTrxId(txId, FIELD_TX_ACTIVE);
    }

    @Override
    public boolean isCommitted(long txId) {
        if (txId == SUPER_TX_ID) {
            return true;
        }
        return checkTrxId(txId, FIELD_TX_COMMITTED);
    }

    @Override
    public boolean isAborted(long txId) {
        if (txId == SUPER_TX_ID) {
            return false;
        }
        return checkTrxId(txId, FIELD_TX_ABORTED);
    }

    @Override
    public void close() {
        try {
            channel.close();
            file.close();
        } catch (IOException e) {
            ExceptionUtils.getMessage(e);
        }
    }
}
